package org.lzfto.flink.demo.business.flinkTest.flinkObjects.source;

import java.util.List;
import java.util.Map;

import org.apache.flink.runtime.checkpoint.Checkpoint;

public class CustEnumeratorCKT implements Checkpoint {
    private final Map<Integer, List<byte[]>> serializedPendingSplits;
    private final List<byte[]> serializedAllSplits;
    private final long checkpointId;

    public CustEnumeratorCKT(Map<Integer, List<byte[]>> serializedPendingSplits,
                                  List<byte[]> serializedAllSplits,
                                  long checkpointId) {
        this.serializedPendingSplits = serializedPendingSplits;
        this.serializedAllSplits = serializedAllSplits;
        this.checkpointId = checkpointId;
    }

    @Override
    public long getCheckpointID() {
        return checkpointId;
    }

    @Override
    public DiscardObject markAsDiscarded() {
        return new DiscardObject() {
			@Override
			public void discard() throws Exception {
				// 释放一些资源，比如数据库连接或缓存
		        serializedPendingSplits.clear();
		        serializedAllSplits.clear();
		        System.out.println("当前分片已经被抛弃");
			}
        };
    }

    public Map<Integer, List<byte[]>> getSerializedPendingSplits() {
        return serializedPendingSplits;
    }

    public List<byte[]> getSerializedAllSplits() {
        return serializedAllSplits;
    }
}